package com.mr.WordCount;

import com.mr.WordCount.Component.WordCountMapper;
import com.mr.WordCount.Component.WordCountReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


/**
 * 组装 MapReduce 作业，并提交到服务器运行
 * (1)hdfs dfs -put wc.txt
 * (2)yarn jar leonard.hadoop.jar com.mr.WordCount.WordCountApp
 * -D input=/user/briup/wc.txt -D output=/user/briup/wcresult
 * (3)hdfs dfs -cat wcresult/part-r-00000
 */
public class WordCountApp extends Configured implements Tool {
    @Override
    public int run(String[] strings) throws Exception {
        //读取配置文件,用来连接hdfs的环境
        Configuration conf = getConf();
        String input = conf.get("input");
        String output = conf.get("output");

        //创建一个Job
        Job job = Job.getInstance();
        //设置运行主类
        job.setJarByClass(WordCountApp.class);
        job.setJobName("wordCountByleonard");

        //设置Mapper和Reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        //设置Mapper输出key和value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //设置Reducer输出key和value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setNumReduceTasks(1);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.addInputPath(job, new Path(input));
        TextOutputFormat.setOutputPath(job, new Path(output));
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        //根据作业结果,终止当前运行的 Java 虚拟机,退出程序
        System.exit(new ToolRunner().run(new WordCountApp(),args));
    }
}
